#include "sd_server.h"
#include "sd_global.h"
#include "sd_ttree.h"
#include "sd_blist.h"
#include "sd_hashtable.h"
#include "sd_memory.h"
#include "sd_log.h"
#include "sd_transaction.h"
#include "sd_oplog.h"
#include "sd_thread.h"

#define SD_MAX_STORE_NUM 50


#define SD_MAX_NUM_TTREE_NODE_RECORDS 64
#define SD_MAX_NUM_ACTIVE_TXN  60
#define SD_LOG_LIST_NODE_SIZE  20
#define SD_LOCKTABLE_HASH_SIZE 17
#define SD_LOCKTABLE_BUCKET_NODE_SIZE 5
#define SD_TXN_LOCK_TIMEOUT_SEC 12000
#define SD_TXN_LOCK_TIMEOUT_NSEC 300


typedef struct sd_Store
{
	char name[SD_MAX_VARCHAR_LEN];
	sd_TTree* ttree;
	sd_KeyType keytype;
}sd_Store;

struct IdxState
{
	sd_Store* store;
	sd_Record currentRecord;
	bool bNewOpened;
};

static sd_Store g_StoreTable[SD_MAX_STORE_NUM];
static int g_NumStore = 0;
static sd_uid g_TreeIDCounter = 0;
static sd_uid g_RIDCounter = 0;
static sd_TxnManager* g_TxnManager = NULL;
static sd_SpinLock g_SpinLock;
static sd_Hashtable* g_pOpenedIndexTable = NULL;

ErrCode create(KeyType type, char *name)
{
	sd_Ret ret;
	sd_Store* newStore = &g_StoreTable[g_NumStore++];
	sd_KeyCompare key_cmp = NULL;
#ifdef WIN32
	switch(type)
	{
		case KEYTYPE_SHORT:
			newStore->keytype = SD_KEYTYPE_SHORT;
			key_cmp = sd_KeyShortCompare;
			break;
		case KEYTYPE_INT:
			newStore->keytype = SD_KEYTYPE_INT;
			key_cmp = sd_KeyIntCompare;
			break;
		case KEYTYPE_VARCHAR:
			newStore->keytype = SD_KEYTYPE_VARCHAR;
			key_cmp = sd_KeyCharCompare;
			break;
		default:
			key_cmp = NULL;
			SD_ASSERT(false);
			break;
	}
#else
	switch(type)
	{
	case SHORT:
		newStore->keytype = SD_KEYTYPE_SHORT;
		key_cmp = sd_KeyShortCompare;
		break;
	case INT:
		newStore->keytype = SD_KEYTYPE_INT;
		key_cmp = sd_KeyIntCompare;
		break;
	case VARCHAR:
		newStore->keytype = SD_KEYTYPE_VARCHAR;
		key_cmp = sd_KeyCharCompare;
		break;
	default:
		key_cmp = NULL;
		SD_ASSERT(false);
		break;
	}
#endif
	strcpy(newStore->name, name);

	/* Create the index tree */
	ret = sd_TTreeCreate(&newStore->ttree, SD_MAX_NUM_TTREE_NODE_RECORDS, key_cmp);
	if (SD_FAILED(ret))
	{
		SD_ASSERT(false);
		return FAILURE;
	}
	newStore->ttree->treeId = g_TreeIDCounter++;

	/* Create the transaction manager */
	if (g_TxnManager == NULL)
	{
		struct timespec ts;
		ts.tv_sec = SD_TXN_LOCK_TIMEOUT_SEC;
		ts.tv_nsec = SD_TXN_LOCK_TIMEOUT_NSEC;
		ret = sd_TxnManagerCreate(&g_TxnManager, 
			SD_MAX_NUM_ACTIVE_TXN, 
			SD_LOG_LIST_NODE_SIZE, 
			SD_LOCKTABLE_HASH_SIZE,
			SD_LOCKTABLE_BUCKET_NODE_SIZE,
			ts);
		if (SD_FAILED(ret))
		{
			SD_ASSERT(false);
			return FAILURE;
		}

		sd_SpinLock_Init(&g_SpinLock);
	}

	if (g_pOpenedIndexTable == NULL)
	{
		size_t addressSize = sizeof(void*);
		if (addressSize == 8)
			ret = sd_HashtableCreate(&g_pOpenedIndexTable,sd_HashFunc_uint64, sd_HashKeyCompare_uint64, 
			23, 16, addressSize, addressSize, NULL);
		else if (addressSize == 4)
			ret = sd_HashtableCreate(&g_pOpenedIndexTable,sd_HashFunc_uint32, sd_HashKeyCompare_uint32, 
			23, 16, addressSize, addressSize, NULL);
		else
			SD_ASSERT(false);

		if (SD_FAILED(ret))
		{
			SD_ASSERT(false);
			return FAILURE;
		}
	}

	return SUCCESS;
}

ErrCode openIndex(const char *name, IdxState **idxState)
{
	int i;
	for (i=0; i<g_NumStore; i++)
	{
		if (strcmp(name, g_StoreTable[i].name) == 0)
		{
			*idxState = (IdxState*)sd_malloc(sizeof(IdxState));
			(*idxState)->store = &g_StoreTable[i];
			(*idxState)->bNewOpened = true;
			sd_SpinLock_Lock(&g_SpinLock);
			sd_HashtableInsert(g_pOpenedIndexTable, idxState, idxState);
			sd_SpinLock_Unlock(&g_SpinLock);
			return SUCCESS;
		}
	}
	return DB_DNE;
}

ErrCode closeIndex(IdxState *idxState)
{
	if (idxState == NULL)
		return FAILURE;
	sd_SpinLock_Lock(&g_SpinLock);
	sd_HashtableDelete(g_pOpenedIndexTable, &idxState);
	sd_SpinLock_Unlock(&g_SpinLock);
	sd_free(idxState);	
	return SUCCESS;
}

ErrCode beginTransaction(TxnState **txn)
{
	sd_TxnState txnState;
	*txn = (TxnState*)sd_malloc(sizeof(sd_TxnState));
	sd_TxnManagerBeginTransaction(g_TxnManager, &txnState);
	txnState.bNewCreated = true;
	memcpy(*txn, &txnState, sizeof(sd_TxnState));
	return SUCCESS;
}

ErrCode abortTransaction(TxnState *txn)
{
	sd_TxnState txnState;
	memcpy(&txnState, txn, sizeof(sd_TxnState));
	sd_TxnManagerAbortTransaction(&txnState);
	sd_free(txn);
	return SUCCESS;
}


ErrCode commitTransaction(TxnState *txn)
{
	sd_TxnState txnState;
	memcpy(&txnState, txn, sizeof(sd_TxnState));
	sd_TxnManagerCommitTransaction(&txnState);
	sd_free(txn); 
	return SUCCESS;
}

ErrCode get(IdxState *idxState, TxnState *txn, Record *record)
{
	sd_Ret ret;
	sd_Store* store = idxState->store;
	sd_Record tempRecord;
	sd_uid prevRID;
	memcpy(&tempRecord.key, &record->key, sizeof(sd_Key));
	tempRecord.payload[0]=0;

_get_begin:
	ret = sd_TTreeSearchRID(store->ttree, &tempRecord, &tempRecord.rid);
	while(ret == SD_RET_OK)
	{
		if (txn)
		{
			prevRID = tempRecord.rid;

			/* S-Lock this record*/
			if(sd_TxnRLock((sd_TxnState*)txn, tempRecord.rid)==SD_RET_DEAD_LOCK)
			{
				/* Wait for another transaction's thread */
				sd_sched_yield();
				sd_log("deadlock %d\n",((sd_TxnState*)txn)->tid);
				return DEADLOCK;
				//goto _get_begin;
			}

			/* Re-search the record.
			Notice that the record may be deleted by other transaction */
			ret = sd_TTreeSearch(store->ttree, &tempRecord.key, &tempRecord);
			if (ret == SD_RET_OK)
			{
				if (prevRID == tempRecord.rid)
				{
					strcpy(record->payload, tempRecord.payload);
					idxState->bNewOpened = false;
					idxState->currentRecord = tempRecord;
					((sd_TxnState*)txn)->bNewCreated = false;
					return SUCCESS;
				}
				else
				{
					/* This record is not the previous locked record */
					continue;
				}
			}
			else if (ret == SD_RET_KEY_NOTFOUND)
			{
				/* Try to find another record with the given key */
				ret = sd_TTreeSearchRID(store->ttree, &tempRecord, &tempRecord.rid);
				continue;
			}
			else
			{
				SD_ASSERT(false);
				return FAILURE;
			}
		}
		else
		{
			memcpy(record->payload, tempRecord.payload, SD_MAX_PAYLOAD_LEN+1);
			idxState->bNewOpened = false;
			idxState->currentRecord = tempRecord;
			return SUCCESS;
		}
	}

	if (ret == SD_RET_KEY_NOTFOUND)
	{
		idxState->bNewOpened = false;
		idxState->currentRecord.key = tempRecord.key;
		idxState->currentRecord.payload[0]=0;
		((sd_TxnState*)txn)->bNewCreated = false;
		return KEY_NOTFOUND;
	}
	else
		return FAILURE;
}

ErrCode getNext(IdxState *idxState, TxnState *txn, Record *record)
{
	sd_Ret ret;
	sd_Store* store = idxState->store;
	sd_Record tempRecord;
	sd_uid prevRID;
	memcpy(&tempRecord.key, &record->key, sizeof(sd_Key));

_getnext_begin:
	if (idxState->bNewOpened || ((sd_TxnState*)txn)->bNewCreated)
	{
		ret = sd_TTreeGetFirst(store->ttree, &tempRecord);
		while(ret == SD_RET_OK)
		{
			if (txn)
			{
				prevRID = tempRecord.rid;
				/* S-Lock this record*/
				
				/* deadlock detection */
				if( sd_TxnRLock((sd_TxnState*)txn, tempRecord.rid) == SD_RET_DEAD_LOCK)
				{
					/* Wait for another transaction's thread */
					sd_sched_yield();
					sd_log("deadlock %d\n",((sd_TxnState*)txn)->tid);
					return DEADLOCK;
					goto _getnext_begin;
				}
				/* deadlock detection */

				ret = sd_TTreeGetFirst(store->ttree, &tempRecord);
				if (ret == SD_RET_OK)
				{
					if (prevRID == tempRecord.rid) /* This record is not the record acquired lock before */
					{
						memcpy(&record->key, &tempRecord.key, sizeof(sd_Key));
						strcpy(record->payload, tempRecord.payload);
						idxState->bNewOpened = false;
						idxState->currentRecord = tempRecord;
						((sd_TxnState*)txn)->bNewCreated = false;
						return SUCCESS;
					}
					else
					{
						/* This record is not the previous locked record */
						continue;
					}
				}
				else if (ret == SD_RET_DB_END)
				{
					return DB_END;
				}
				else
				{
					SD_ASSERT(false);
					return FAILURE;
				}
				
			}
			else
			{
				memcpy(&record->key, &tempRecord.key, sizeof(sd_Key));
				strcpy(record->payload, tempRecord.payload);
				idxState->bNewOpened = false;
				idxState->currentRecord = tempRecord;
				return SUCCESS;
			}
		}
		if (ret == SD_RET_DB_END)
			return DB_END;
		else
			return FAILURE;
	}
	else
	{
		ret = sd_TTreeSearchNext(store->ttree, &idxState->currentRecord, &tempRecord);
		while (ret == SD_RET_OK)
		{
			if (txn)
			{
				prevRID = tempRecord.rid;
				/* S-Lock this record*/
				
				/*deadlock detection*/
				if(sd_TxnRLock((sd_TxnState*)txn, tempRecord.rid)==SD_RET_DEAD_LOCK)
				{
					/* Wait for another transaction's thread */
					sd_sched_yield();
					sd_log("deadlock %d\n",((sd_TxnState*)txn)->tid);
					return DEADLOCK;
					//goto _getnext_begin;
				}

				ret = sd_TTreeSearchNext(store->ttree, &idxState->currentRecord, &tempRecord);
				if (ret == SD_RET_OK)
				{
					if (prevRID == tempRecord.rid) /* This record is not the record acquired lock before */
					{
						memcpy(&record->key, &tempRecord.key, sizeof(sd_Key));
						strcpy(record->payload, tempRecord.payload);
						idxState->currentRecord = tempRecord;
						return SUCCESS;
					}
					else
					{
						/* This record is not the previous locked record */
						continue;
					}
				}
				else if (ret == SD_RET_DB_END)
				{
					return DB_END;
				}
				else
				{
					SD_ASSERT(false);
					return FAILURE;
				}
			}
			else
			{
				memcpy(&record->key, &tempRecord.key, sizeof(sd_Key));
				strcpy(record->payload, tempRecord.payload);
				idxState->currentRecord = tempRecord;
				return SUCCESS;
			}
		}
		if (ret == SD_RET_DB_END)
			return DB_END;
		else
			return FAILURE;
	}
}

ErrCode insertRecord(IdxState *idxState, TxnState *txn, Key *k, const char* payload)
{
	sd_Ret ret;
	sd_Store* store = idxState->store;
	sd_Record record;
	
	memcpy(&record.key, k, sizeof(sd_Key));
	memcpy(record.payload,payload, SD_MAX_PAYLOAD_LEN + 1);
	
	sd_SpinLock_Lock(&g_SpinLock);
	record.rid = g_RIDCounter;
	SD_UID_INC(g_RIDCounter);
	sd_SpinLock_Unlock(&g_SpinLock);

	if (txn)
	{
		
		/* X-Lock this record*/

		/*deadlock detection*/
		if( sd_TxnWLock((sd_TxnState*)txn, record.rid)==SD_RET_DEAD_LOCK)
		{
			/* Wait for another transaction's thread */
			sd_sched_yield();
			sd_log("deadlock %d\n", ((sd_TxnState*)txn)->tid);
			return DEADLOCK;
		}
		/* Add this operation into transaction's log */
		sd_OpLogItem logItem;
		logItem.type = SD_OP_INSERT;
		logItem.record = record;
		logItem.index = store->ttree;
		sd_TxnAddLog((sd_TxnState*)txn, &logItem);
	}

	ret = sd_TTreeInsert(store->ttree, &record);
	if (ret == SD_RET_OK)
		return SUCCESS;
	else if (ret == SD_SET_RECORD_EXISTS)
		return ENTRY_EXISTS;
	else 
		return FAILURE;
}

ErrCode deleteRecord(IdxState *idxState, TxnState *txn, Record *record)
{
	sd_Ret ret;
	sd_Store* store = idxState->store;
	sd_Record tempRecord;
	sd_uid prevRID;
	uint32 numDeleted = 0;
	bool bContainPayload = record->payload[0] != 0;
	sd_OpLogItem logItem;
	logItem.type = SD_OP_DELETE;
	logItem.index = store->ttree;

	memcpy(&tempRecord.key, &(record->key), sizeof(sd_Key));
	strcpy(tempRecord.payload, record->payload);

_delete_begin:
	ret = sd_TTreeSearchRID(store->ttree, &tempRecord, &tempRecord.rid);
	if (ret == SD_RET_OK)
	{
		/* Add this operation into transaction's log */
		if (txn)
		{
_delete_relock:
			prevRID = tempRecord.rid;
			/* deadlock detection */
			if(sd_TxnWLock((sd_TxnState*)txn, tempRecord.rid)==SD_RET_DEAD_LOCK)
			{
				/* Wait for another transaction's thread */
				sd_sched_yield();
				sd_log("deadlock\n");
				return DEADLOCK;
				//goto _delete_begin;
			}
			
			ret = sd_TTreeSearchRID(store->ttree, &tempRecord, &tempRecord.rid);
			if (ret == SD_RET_OK)
			{
				if (prevRID == tempRecord.rid)
				{
					ret = sd_TTreeDelete(store->ttree, &tempRecord);
					SD_ASSERT(ret == SD_RET_OK);
					logItem.record = tempRecord;
					sd_TxnAddLog((sd_TxnState*)txn, &logItem);
					numDeleted++;
					if (bContainPayload == false)
					{
						tempRecord.payload[0] = 0;
						goto _delete_begin;
					}
				}
				else
				{
					/* This record is not the previous locked record */
					goto _delete_relock;
				}
			}
			else if (ret == SD_RET_KEY_NOTFOUND)
			{
				if (bContainPayload == false)
				{
					/* Zero the payload */
					tempRecord.payload[0] = 0;
					goto _delete_begin;
				}
			}
			else
			{
				SD_ASSERT(false);
				return FAILURE;
			}
		}
		else
		{
			ret = sd_TTreeDelete(store->ttree, &tempRecord);
			if (ret == SD_RET_OK)
				numDeleted++;
			if (bContainPayload == false)
			{
				/* Zero the payload */
				tempRecord.payload[0] = 0;
				goto _delete_begin;
			}
		}
	}
	if (numDeleted == 0)
		return KEY_NOTFOUND;
	else
		return SUCCESS;
}

void ExitAll()
{
	int i;
	sd_HashtableIterator iter;
	sd_HashtableBegin(g_pOpenedIndexTable, &iter);
	while(sd_HashtableNext(&iter) != SD_RET_FALSE)
	{
		void* openedIndex;
		sd_HashtableGet(&iter, &openedIndex);
		sd_free(openedIndex);
	}
	sd_HashtableClear(g_pOpenedIndexTable);
	sd_HashtableFree(g_pOpenedIndexTable);
	g_pOpenedIndexTable = NULL;

	for (i=0; i<g_NumStore; i++)
	{
		sd_Store* newStore = &g_StoreTable[i];
		sd_TTreeFree(newStore->ttree);
	}
	if (g_TxnManager != NULL)
	{
		sd_TxnManagerFree(g_TxnManager);
		g_TxnManager = NULL;
		sd_SpinLock_Destroy(&g_SpinLock);
	}
	g_NumStore = 0;
}

void PrintIdxState(IdxState* idx, FILE* fp)
{
	sd_TTreePrint(idx->store->ttree, fp);
}
